package com.qf.kafka.spring.boot.demo.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class MyConsumer {

 /* @KafkaListener(topics = "my-replicated-topic",groupId = "MyGroup1")
  public void listenGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
    String value = record.value();
    System.out.println(value);
    System.out.println(record);
    //手动提交offset
    ack.acknowledge();
  }*/

  @KafkaListener(
          groupId = "testGroup", // 消费组 ID
          topicPartitions = {    // 监听主题和分区
                  @TopicPartition(
                          topic = "my-replicated-topic", // 改成和生产者一致
                          partitions = {"0"}             // 指定分区 0
                  )
          },
          concurrency = "1" ,// 同组内并发消费者数量，不超过分区数
          containerFactory = "kafkaListenerContainerFactory" // 使用自定义工厂
  )
  public void listenMyTopic(ConsumerRecord<String, String> record, Acknowledgment ack) {
      if (record.value() == null) {
          System.out.println("收到空消息，跳过处理");
          return;
      }
    // 获取消息内容
    String value = record.value();
    System.out.println("收到消息：" + value);

    // 打印完整消息信息
    System.out.println(record);

    // 手动提交 offset
    ack.acknowledge();
  }



//  // 使用 @KafkaListener 注解监听 Kafka 消息
//  @KafkaListener(
//          groupId = "testGroup", // 指定消费者所属的消费组 ID，同组内的消费者会均摊分区
//          topicPartitions = {    // 指定要监听的主题和分区信息
//                  @TopicPartition(
//                          topic = "topic1",         // 监听的主题名
//                          partitions = {"0", "1"}   // 指定监听 topic1 的 0、1 分区
//                  ),
//                  @TopicPartition(
//                          topic = "topic2",         // 监听的另一个主题 topic2
//                          partitions = "0",         // 默认监听 topic2 的 0 分区
//                          partitionOffsets = @PartitionOffset(
//                                  partition = "1",      // 针对 topic2 的分区 1
//                                  initialOffset = "100" // 指定从偏移量 100 开始消费
//                          )
//                  )
//          },
//          concurrency = "3" // 同一消费组下的并发消费者线程数，建议不要超过分区总数
//  )
//  public void listenGroupPro(ConsumerRecord<String, String> record, Acknowledgment ack) {
//    // ConsumerRecord 封装了 Kafka 消息的所有信息（topic、partition、offset、key、value 等）
//    // 获取消息的值（payload）
//    String value = record.value();
//    // 打印消息内容
//    System.out.println(value);
//    // 打印完整的消息信息，包括分区、偏移量、key 等
//    System.out.println(record);
//
//    // 手动提交 offset，告诉 Kafka 该消息已经消费成功
//    // Acknowledgment 对象只有在手动 ack 时才有用
//    ack.acknowledge();
//  }



  /*@KafkaListener(topics = "my-replicated-topic",groupId = "MyGroup2")
  public void listensGroup(ConsumerRecords<String, String> records, Acknowledgment ack) {
    String value = record.value();
    System.out.println(value);
    System.out.println(record);
    //手动提交offset
    ack.acknowledge();
  }*/

}
